package com.itheima.kafkaTest;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.servlet.http.HttpServletRequest;
import javax.xml.ws.spi.http.HttpHandler;

/**
 * @PackageName: com.itheima.kafkaTest
 * @ClassName: KafkaTest
 * @Author: Renyz
 * @Date: 2021/12/9 11:24
 * @Description: //TODO
 */



@RestController
@RequestMapping("/kafka")
public class KafkaTest {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;


    // 发送消息
    @GetMapping("/normal/{message}")
    public void sendMessage1(@PathVariable("message") String normalMessage, HttpServletRequest handler) {
        System.out.println(normalMessage);

        kafkaTemplate.send("test", normalMessage).addCallback(failure -> {
            System.out.println(failure.getRecordMetadata().topic());
            System.out.println(failure.getRecordMetadata().partition());
            System.out.println(failure.getRecordMetadata().offset());
        },success ->{
            System.out.println(success.getMessage());
        });

    }
    // 消费监听
    @KafkaListener(topics = {"test"})
    public void onMessage1(ConsumerRecord<?, ?> record){
        // 消费的哪个topic、partition的消息,打印出消息内容

//        System.out.println(record.partition());
//        System.out.println(record.headers());
//        System.out.println(record.offset());
//        System.out.println(record.serializedKeySize());
//        System.out.println(record.timestamp());
//        System.out.println(record.partition());

        System.out.println("简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
    }


}
